package com.wanke.util;

import com.wanke.constant.InfluxDBConsts;
import com.wanke.domain.po.Measurement;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.TimeUtil;

import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * InfluxDB数据库工具类
 * todo 该类只能使用数据库默认的保存策略.尚未查明原因
 *
 * @author liuronglei
 */
public class InfluxDBUtil {

    private InfluxDBConnection influxDBConnection;


    public InfluxDBUtil(InfluxDBConnection influxDBConnection) {
        this.influxDBConnection = influxDBConnection;
    }

    private static final Map<Long, String> pidMap = new ConcurrentHashMap<>();


    /**
     * 字段转义
     *
     * @param column 字段
     * @return String
     */
    private String setColumns(String column) {
        String[] cols = column.split("_");
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < cols.length; i++) {
            String col = cols[i].toLowerCase();
            if (i != 0) {
                String start = col.substring(0, 1).toUpperCase();
                String end = col.substring(1).toLowerCase();
                col = start + end;
            }
            sb.append(col);
        }
        return sb.toString();
    }


    /**
     * 获得自某个时间点之前的量测
     *
     * @param dtime 时间
     * @return Measurement
     */
    public List<Measurement> getTempRowData(Date dtime, long size, int page) {
        String sql = String.format("select * from %s where   time <='%s' limit  %d offset  %d ",
                InfluxDBConsts.MEASUREMENT_TEMP, TimeUtil.toInfluxDBTimeFormat(dtime.getTime()), size, page);
        QueryResult results = influxDBConnection.query(sql);
        if (results.getResults().get(0).getSeries() == null) {
            return Collections.emptyList();
        }
        QueryResult.Series series = results.getResults().get(0).getSeries().get(0);
        List<List<Object>> values = series.getValues();
        List<Measurement> result = new ArrayList<>();
        for (List<Object> value : values) {
            Measurement measurement = new Measurement();
            Date date = new Date(TimeUtil.fromInfluxDBTimeFormat((String) value.get(0)));
            String format = DateTimeUtils.format(date);
            measurement.setDtime(format);
            measurement.setFlag(((Double) value.get(1)).intValue());
            measurement.setPointNumber(Long.parseLong((String) value.get(2)));
            measurement.setVal((Double) value.get(3));
            measurement.setRtime(date.getTime());
            result.add(measurement);
        }
        return result;
    }

    /**
     *  获得备份表数据
     * @param start
     * @param end
     * @param size
     * @param page
     * @return
     */
    public List<Measurement> getBackRowData(Date start, Date end, long size, int page) {
        String sql = String.format("select * from %s where   time >='%s'  and time <='%s' limit %d offset %d ",
                InfluxDBConsts.MEASUREMENT_BACK, TimeUtil.toInfluxDBTimeFormat(start.getTime()), TimeUtil.toInfluxDBTimeFormat(end.getTime()), size, page);
        QueryResult results = influxDBConnection.query(sql);
        if (results.getResults().get(0).getSeries() == null) {
            return Collections.emptyList();
        }
        QueryResult.Series series = results.getResults().get(0).getSeries().get(0);
        List<List<Object>> values = series.getValues();
        List<Measurement> result = new ArrayList<>();
        for (List<Object> value : values) {
            Measurement measurement = new Measurement();
            Date date = new Date(TimeUtil.fromInfluxDBTimeFormat((String) value.get(0)));
            String format = DateTimeUtils.format(date);
            measurement.setDtime(format);
            measurement.setFlag(((Double) value.get(1)).intValue());
            measurement.setPointNumber(Long.parseLong((String) value.get(2)));
            measurement.setVal((Double) value.get(3));
            measurement.setRtime(date.getTime());
            result.add(measurement);
        }
        return result;
    }

    /**
     * 为表添加 or 更新数据（批量）
     *
     * @param measList 量测对象列表
     */
    public void putTempData(List<Measurement> measList) {
        List<String> records = new ArrayList<>();
        for (Measurement item : measList) {
            Map<String, String> tags = new HashMap<>();
            if (item.getPointNumber() == null || item.getVal() == null) {
                continue;
            }
            tags.put("pid", String.valueOf(item.getPointNumber()));
            Map<String, Object> fields = new HashMap<>();
            fields.put("val", item.getVal());
            fields.put("flag", item.getFlag() == null? 0:item.getFlag());
            // 时间使用毫秒为单位
            Point point = influxDBConnection.pointBuilder(InfluxDBConsts.MEASUREMENT_TEMP, item.getRtime(), tags, fields);
            BatchPoints batchPoints = BatchPoints.database(InfluxDBConsts.DATABASE).tag("pid", String.valueOf(item.getPointNumber()))
                    .retentionPolicy(InfluxDBConsts.RETENTION_POLICY).consistency(InfluxDB.ConsistencyLevel.ALL).build();
            batchPoints.point(point);
            records.add(batchPoints.lineProtocol());
        }
        influxDBConnection.batchInsert(InfluxDBConsts.DATABASE, InfluxDBConsts.RETENTION_POLICY, InfluxDB.ConsistencyLevel.ALL, records);
    }

    /**
     * 为备份表表添加 or 更新数据（批量）
     *
     * @param measList 量测对象列表
     */
    public void putBackData(List<Measurement> measList) {
        List<String> records = new ArrayList<>();
        for (Measurement item : measList) {
            Map<String, String> tags = new HashMap<>();
            if (item.getPointNumber() == null || item.getVal() == null) {
                continue;
            }
            tags.put("pid", String.valueOf(item.getPointNumber()));
            Map<String, Object> fields = new HashMap<>();
            fields.put("val", item.getVal());
            fields.put("flag", item.getFlag() == null? 0:item.getFlag());
            // 时间使用毫秒为单位
            Point point = influxDBConnection.pointBuilder(InfluxDBConsts.MEASUREMENT_BACK, item.getRtime(), tags, fields);
            BatchPoints batchPoints = BatchPoints.database(InfluxDBConsts.DATABASE).tag("pid", String.valueOf(item.getPointNumber()))
                    .retentionPolicy(InfluxDBConsts.RETENTION_POLICY).consistency(InfluxDB.ConsistencyLevel.ALL).build();
            batchPoints.point(point);
            records.add(batchPoints.lineProtocol());
        }
        influxDBConnection.batchInsert(InfluxDBConsts.DATABASE, InfluxDBConsts.RETENTION_POLICY, InfluxDB.ConsistencyLevel.ALL, records);
    }


    /**
     * 获得数据ID在时间区间内的量测总数量
     *
     * @return Measurement
     * @param time
     */
    public Long getTempCount(Date time) {
        StringBuilder sqlBuilder = new StringBuilder(String.format("select count(val) from %s where time < '%s'", InfluxDBConsts.MEASUREMENT_TEMP, TimeUtil.toInfluxDBTimeFormat(time.getTime())));
        QueryResult results = influxDBConnection.query(sqlBuilder.toString());
        if (results.getResults().get(0).getSeries() == null) {
            return null;
        }
        QueryResult.Series serie = results.getResults().get(0).getSeries().get(0);
        List<String> columns = serie.getColumns();
        List<List<Object>> values = serie.getValues();
        if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
            return null;
        }
        List<Object> list = values.get(0);
        if (list != null && list.size() > 1) {
            return ((Double) list.get(1)).longValue();
        } else {
            return 0L;
        }
    }


    /**
     * 获得数据ID在时间区间内的量测总数量
     *
     * @return Measurement
     */
    public Long getMeasurementBackCount(Date start, Date end) {
        StringBuilder sqlBuilder = new StringBuilder(String.format("select count(val) from %s where time >= '%s' and time <= '%s'",
                InfluxDBConsts.MEASUREMENT_BACK, TimeUtil.toInfluxDBTimeFormat(start.getTime()), TimeUtil.toInfluxDBTimeFormat(end.getTime())));
        QueryResult results = influxDBConnection.query(sqlBuilder.toString());
        if (results.getResults().get(0).getSeries() == null) {
            return null;
        }
        QueryResult.Series serie = results.getResults().get(0).getSeries().get(0);
        List<String> columns = serie.getColumns();
        List<List<Object>> values = serie.getValues();
        if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
            return null;
        }
        List<Object> list = values.get(0);
        if (list != null && list.size() > 1) {
            return ((Double) list.get(1)).longValue();
        } else {
            return 0L;
        }
    }

    public boolean deleteTempRow(Date dtime) {
        String sql = String.format("delete from %s where time < '%s'",
                InfluxDBConsts.MEASUREMENT_TEMP, TimeUtil.toInfluxDBTimeFormat(dtime.getTime()), ZoneId.systemDefault());
        String error = influxDBConnection.deleteMeasurementData(sql);
        return StringUtil.isEmpty(error);
    }

}
